﻿import threading
import select
import _socket as socket

class AsyncManager(object):
    
    def __init__(self, handler):
        def null(sock, errno): pass
        self.do_accept = getattr(handler, 'do_accept', null)
        self.do_connect = getattr(handler, 'do_connect', null)
        self.do_read = getattr(handler, 'do_read', null)
        self.do_write = getattr(handler, 'do_write', null)
        self.do_close = getattr(handler, 'do_close', null)
        self._csr = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self._csr.bind(('127.0.0.1', 0))
        self._csw = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self._csw.connect(self._csr.getsockname())
        self._ifds = set()
        self._rfds = {self._csr}
        self._wfds = set()
        self._xfds = set()
        self._thrd = threading.Thread(target=self._async_proc)
        self._thrd.start()
    
    def __del__(self):
        self._csw.send('q')
        self._thrd.join()
        self._csw.close()
        self._csr.close()
        self._thrd = None
        self._ifds = None
        self._rfds = None
        self._wfds = None
        self._xfds = None
        self.do_accept = None
        self.do_connect = None
        self.do_read = None
        self.do_write = None
        self.do_close = None
    
    def set_connecting(self, sock):
        self._ifds.add(sock)
        self._wfds.add(sock)
        self._xfds.add(sock)
        self._csw.send('.')
    
    def set_listening(self, sock):
        self._ifds.add(sock)
        self._rfds.add(sock)
        self._csw.send('.')
    
    def set_readable(self, sock, flag):
        if flag:
            self._rfds.add(sock)
        else:
            self._rfds.discard(sock)
        if threading.current_thread() != self._thrd:
            self._csw.send('.')
    
    def set_writable(self, sock, flag):
        if flag:
            self._wfds.add(sock)
        else:
            self._wfds.discard(sock)
        if threading.current_thread() != self._thrd:
            self._csw.send('.')

    def _async_proc(self):
        print "starting async_proc"
        while True:
            print 'before select:', len(self._rfds), len(self._wfds), len(self._xfds)
            rfds, wfds, xfds = select.select(self._rfds, self._wfds, self._xfds)
            print 'after select:', len(rfds), len(wfds), len(xfds)
            for sock in wfds:
                if sock in self._ifds:
                    self._ifds.remove(sock)
                    self._wfds.remove(sock)
                    self._xfds.remove(sock)
                    self.do_connect(sock, 0)
                else:
                    self.do_write(sock, 0)
            for sock in xfds:
                if sock in self._ifds:
                    self._ifds.remove(sock)
                    self._wfds.remove(sock)
                    self._xfds.remove(sock)
                    errno = sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
                    self.do_connect(sock, errno)
                else:
                    self.do_read(sock, socket.MSG_OOB)
            for sock in rfds:
                if sock is self._csr:
                    val = sock.recv(1)
                    print "csr.recv() ->", val
                    if val == 'q':
                        break
                elif sock in self._ifds:
                    self.do_accept(sock, 0)
                else:
                    try:
                        avail = sock.recv(1, socket.MSG_PEEK)
                        errno = 0
                    except socket.error as err:
                        avail = False
                        errno = err.errno
                    if avail:
                        self.do_read(sock, 0)
                    else:
                        self._rfds.remove(sock)
                        self.do_close(sock, errno)
        print "stopping async_proc"

class tester:
    def __init__(self):
        self.am = AsyncManager(self)
    def do_accept(self, sock, errno):
        print "do_accept()", sock, errno
    def do_connect(self, sock, errno):
        print "do_connect()", sock, errno
    def do_read(self, sock, errno):
        print "do_read()", sock, errno
        self.am.set_readable(sock, False)
    def do_write(self, sock, errno):
        print "do_write()", sock, errno
        self.am.set_writable(sock, False)
    def do_close(self, sock, errno):
        print "do_close()", sock, errno
